package edu.pku.hawk.controller.util;

import java.io.InputStream;
import java.io.OutputStream;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

/*
 * Only for single thread!
 */
public class MQManager {

	private HashMap<String, MQueue> qs;
	private Connection connection;
	private Session session;

	public MQManager(String user, String password, String url) throws Exception {
		qs = new HashMap<String, MQueue>();

		// Create the connection.
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
		connection = connectionFactory.createConnection();
		connection.start();

		// Create the session
		session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	}

	public MProducerQueue getProducerQueue(String queueName) throws JMSException {
		MQueue q;

		if ((q = qs.get(queueName)) != null) {
			return (MProducerQueue) q;

		} else {
			Destination destination = session.createQueue(queueName);
			MessageProducer producer = session.createProducer(destination);
			producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
			MProducerQueue queue = new MProducerQueue(queueName, session, producer);
			qs.put(queueName, queue);
			return queue;
		}
	}

	public MConsumerQueue getConsumerQueue(String queueName) throws JMSException {
		MQueue q;

		if ((q = qs.get(queueName)) != null) {
			return (MConsumerQueue) q;

		} else {
			Destination destination = session.createQueue(queueName);
			MessageConsumer consumer = session.createConsumer(destination);
			MConsumerQueue queue = new MConsumerQueue(queueName, session, consumer);
			qs.put(queueName, queue);
			return queue;
		}
	}

	public MInputStreamQueue getInputStreamQueue(String queueName) throws JMSException {
		MQueue q;

		if ((q = qs.get(queueName)) != null) {
			return (MInputStreamQueue) q;

		} else {
			Destination destination = session.createQueue(queueName);
			InputStream is = ((ActiveMQConnection)connection).createInputStream(destination);
			MInputStreamQueue queue = new MInputStreamQueue(queueName, is);
			qs.put(queueName, queue);
			return queue;
		}
	}

	public MOutputStreamQueue getOutputStreamQueue(String queueName) throws JMSException {
		MQueue q;

		if ((q = qs.get(queueName)) != null) {
			return (MOutputStreamQueue) q;

		} else {
			Destination destination = session.createQueue(queueName);
			OutputStream is = ((ActiveMQConnection)connection).createOutputStream(destination);
			MOutputStreamQueue queue = new MOutputStreamQueue(queueName, is);
			qs.put(queueName, queue);
			return queue;
		}
	}
	
	public void createProducerQueue(String queueName) throws JMSException {
		getProducerQueue(queueName);
	}

	public void createConsumerQueue(String queueName) throws JMSException {
		getConsumerQueue(queueName);
	}

	public void Finalize() {
		try {
			close();
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}

	public void close() throws JMSException {
		for (Iterator<Map.Entry<String, MQueue>> it = qs.entrySet().iterator(); it.hasNext();) {
			MQueue queue = it.next().getValue();
			if (queue != null) {
				queue.close();
			}
		}
		if (session != null)
			session.close();
		if (connection != null)
			connection.close();
	}

}
